package SparkStreaming

import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Durations, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

object UpdateStateByKey {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setMaster("local[2]")
    conf.setAppName("UpdateStateByKey")

    val sc = new SparkContext()
    //new Streaming有两种方式，若使用第一种方式，则上方不需要再初始化SparkContext
    //在JYM中已经创建了SparkContext
    val ssc = new StreamingContext(conf, Durations.seconds(5))
    ssc.sparkContext.setLogLevel("Error")
    //val ssc = new StreamingContext(sc,Durations.seconds(5))
    //可通过ssc.sparkContext获取到SparkContext的值

    val lines: ReceiverInputDStream[String] = ssc.socketTextStream("hostname", 9000)
    val words: DStream[String] = lines.flatMap(one => {
      one.split(" ")
    })
    val pairsWords: DStream[(String, Int)] = words.map(one => {
      (one, 1)
    })

    /**
      * 根据key更状态，需要设置checkpoint来保存状态
      * 默认key的状态在内存中有一份，在checkpoint目录中有一份
      *
      * 多久会将内存中的数据（每一个key多对应的状态）写入到磁盘一份呢？
      * 如果batchInterval小于10s，那么10s会将内存中的数据写入到磁盘一份
      * 如果batchInterval大于10s，那么就以batchInterval为准
      *
      * 目的：为了防止频繁的HDFS
      * 设置checkpoint两种方式都可以
      */
    ssc.checkpoint("D:/spark")
    //ssc.sparkContext.setCheckpointDir("D:/spark")

    /**
      * currentValues：当前批次某个key对应所有的value组成的一个集合
      * preValue：以往批次当前Key，对应的总状态值
      */
    val result: DStream[(String, Int)] = pairsWords.updateStateByKey((currentValues: Seq[Int], preValue: Option[Int]) => {
      var totalValues = 0
      if (!preValue.isEmpty) {
        totalValues += preValue.get
      }
      for (value <- currentValues) {
        totalValues += value
      }
      Option(totalValues)
    })

    ssc.start()
    ssc.awaitTermination()
    //ssc.stop(true)会清空SparkContext对象
    //ssc.stop(false)则不会清空对象
    ssc.stop()
  }
}
